package com.devilvan.hadoop;

import com.devilvan.map.KeyValueTextMapper;
import com.devilvan.reduce.KeyValueTextReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCountByKeyValueTextInput extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        /***
         * 第一步：读取文件，解析成key,value对，k1   v1
         * 第二步：自定义map逻辑，接受k1   v1  转换成为新的k2   v2输出
         * 第三步：分区。相同key的数据发送到同一个reduce里面去，key合并，value形成一个集合
         * 第四步：排序   对key2进行排序。字典顺序排序
         * 第五步：规约  combiner过程  调优步骤 可选
         * 第六步：分组
         * 第七步：自定义reduce逻辑接受k2   v2  转换成为新的k3   v3输出
         * 第八步：输出k3  v3 进行保存
         */
        Configuration conf = super.getConf();
        // 定义分隔符
        conf.set("key.value.separator.in.input.line","@zolen@");
        // 获取Job实例
        Job job = Job.getInstance(conf, this.getClass().getSimpleName());
        // 若输出路径存在则覆盖
        FileSystem fileSystem = FileSystem.get(conf);
        if (fileSystem.exists(new Path(args[1]))) {
            fileSystem.delete(new Path(args[1]), true);
        }
        job.setJarByClass(this.getClass());
        // 第一步：使用KeyValueTextInputFormat 读取文件
        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.addInputPath(job,new Path(args[0]));
        // 第二步：配置自定义Mapper逻辑，并设置key、value类型
        job.setMapperClass(KeyValueTextMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        // 第三、四、五、六步使用默认配置
        // 第七步：配置自定义Reducer逻辑，并设置key、value类型
        job.setReducerClass(KeyValueTextReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        // 第八步：通过TextOutputFormat 将reduce的结果输出
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(args[1]));
        // 设置reduce任务个数（最终生成文件时按照该数量生成）
        job.setNumReduceTasks(Integer.parseInt(args[2]));
        boolean flag = job.waitForCompletion(true);
        return flag ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            // ToolRunner执行Configured子类的run方法（以上方法）
            int run = ToolRunner.run(conf,new WordCountByKeyValueTextInput(),args);
            // 根据执行返回码进行关闭程序（0
            System.exit(run);
    }
}
